package it.wyx.kafkaclients.producer;

import java.util.Properties;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

/**
 * 带回调函数的异步发送-demo
 * @author wyx
 * @version V 1
 * @Date 2022/10/30
 */
public class ProducerAsyncCallback {
	private static final String BROKER_LIST = "localhost:9092";
	private static final String TOPIC = "test-topic";

	public static void main(String[] args) {
		// 1. 创建 kafka 生产者的配置对象
		Properties properties = new Properties();

		// 2. 给 kafka 配置对象添加配置信息:
		// 连接地址
		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
		// 键、值的序列化方式
		properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
		properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

		// 3、创建 kafka 生产者对象
		KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

		// 4. 调用 send 方法,发送消息
		for (int i = 0; i < 5; i++) {
			producer.send(new ProducerRecord<>(TOPIC, "hello kafka " + i), new Callback() {
				@Override
				public void onCompletion(RecordMetadata metadata, Exception exception) {
					if (exception == null) {
						// 没有异常,输出信息到控制台
						System.out.println("主题：" + metadata.topic() + "->" + "分区：" + metadata.partition());
					} else {
						// 出现异常打印
						exception.printStackTrace();
					}
				}
			});
		}


		//全局事务，保证事务的原子性，一个事务里的顺序执行，要么全部成功，要么全部失败
		try {
			producer.beginTransaction();
			producer.send(new ProducerRecord<>("topic1", "key1", "value1"));
			producer.send(new ProducerRecord<>("topic2", "key2", "value2"));
			producer.commitTransaction(); // 原子提交
		} catch (Exception e) {
			producer.abortTransaction(); // 回滚
		} finally {
			// 5、关闭资源
			producer.close();
		}

	}
}